package com.st.kafkademon;

import java.util.Properties;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

/**
 * @author wuxianwei on 15/11/19.
 *
 *         ps:producer 和 keyedMessage 的泛型必须是一样的 其中 K是记录的键值,V是发送的消息
 */
public class KafkaProducer {

  private final Producer<Integer, String> producer;

  public KafkaProducer(String kafkaServerURL, int kafkaServerPort) {
    Properties props = new Properties();
    props.put("serializer.class", "kafka.serializer.StringEncoder");
    props.put("metadata.broker.list", kafkaServerURL + ":" + kafkaServerPort);
    producer = new Producer<Integer, String>(new ProducerConfig(props));
  }

  public void sendMessage(String message, String topic) {
    System.out.println("message: " + message + ",topic: " + topic);
    producer.send(new KeyedMessage<Integer, String>(topic, message));
  }

}
